package com.shujia.flink.tf

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo2FlatMap {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)


    //scala api
    val scalaWonrdDS: DataStream[String] = lineDS.flatMap(_.split(","))


    //java api
    val javaWordsDS: DataStream[String] = lineDS.flatMap(new FlatMapFunction[String, String] {
      /**
        * 一行数据执行一次，每执行一次collect方法向下游发送一条数据
        *
        * @param value ：一行数据
        * @param out   ： 用于向下游发送护具
        */
      override def flatMap(value: String, out: Collector[String]): Unit = {
        value
          .split(",")
          .foreach(word => {

            //将数据发送到下游
            out.collect(word)
          })
      }
    })

    javaWordsDS.print()

    env.execute()
  }

}
